package com.dk.logc.config;

import com.dk.logc.domain.*;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author 林道敬
 * @Date 2024-01-09
 * @Description
 **/
@Component
public  class GlobalConfig {

    public static final Map<String,Class<?>> typeCategoryMap = new HashMap<>();

    static {
        typeCategoryMap.put("chat", FlogChat.class);
        typeCategoryMap.put("entry", FlogEntry.class);
        typeCategoryMap.put("item", FlogItem.class);
        typeCategoryMap.put("login", FlogLogin.class);
        typeCategoryMap.put("mission", FlogMission.class);
        typeCategoryMap.put("payment",FlogPayment.class);
        typeCategoryMap.put("player_events",FlogPlayerEvents.class);
        typeCategoryMap.put("role_info",FlogRoleInfo.class);
        typeCategoryMap.put("virtual_currency",FlogVirtualCurrency.class);
        typeCategoryMap.put("virtual_resource",FlogVirtualResource.class);
    }

    /**
     *  /**
     *     * 写在接收kafka消息的 同一个类中就行了
     *     * 原先拉去数据是单个json 现在会是多个 list 我设置factory.setConcurrency(1)为一条，
     *     * 但返回数据还是list。这里的 1条是你提交1次的数据。你一次提交了20条也是一个json里的所有
     *     * 算做一条数据。
     *     */
    @Bean
    public KafkaListenerContainerFactory<?> batchFactory(ConsumerFactory consumerFactory){
        ConcurrentKafkaListenerContainerFactory<Integer,String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        // 消费几条数据
        factory.setConcurrency(1);
        factory.getContainerProperties().setPollTimeout(15000);
        //设置为批量消费，每个批次数量在Kafka配置参数中设置
        factory.setBatchListener(true);
        //设置手动提交ackMode
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }

}
